package com.homedo.microservice.open.service.gateway.manager.web.task;


import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.homedo.common.bean.data.RequestData;
import com.homedo.common.bean.model.base.BaseAMO;
import com.homedo.common.bean.vo.Pagination;
import com.homedo.common.dao.conf.DataSourceChooseSupportHolder;
import com.homedo.common.util.Detect;
import com.homedo.common.util.HttpUtility;
import com.homedo.microservice.open.service.gateway.manager.web.bean.po.ServiceGateway;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.bean.po.enhanced.RouteEnhanced;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.bean.po.enhanced.RouteVersionEnhanced;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.bean.query.RouteQuery;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.bean.query.RouteVersionQuery;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.service.RouteService;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.service.RouteServiceService;
import com.homedo.microservice.open.service.gateway.manager.web.gateway.service.RouteVersionService;
import com.homedo.microservice.open.service.gateway.manager.web.persistence.dao.ServiceGatewayDao;
import com.netflix.config.DynamicBooleanProperty;
import com.netflix.config.DynamicIntProperty;
import com.netflix.config.DynamicPropertyFactory;
import com.netflix.config.DynamicStringProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URL;
import java.net.URLConnection;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * 动态监控各路由当前并发数
 */
@Component
public class TurbineConcurrentThreadsNoticeCDTask {
    @Autowired
    private ServiceGatewayDao serviceGatewayDao;
    @Autowired
    private RouteServiceService routeServiceService;
    @Autowired
    private RouteVersionService routeVersionService;
    @Autowired
    private RouteService routeService;

    private static final Logger LOGGER = LoggerFactory.getLogger(TurbineConcurrentThreadsNoticeCDTask.class);

    private DynamicStringProperty hystrixMonitorUrl = DynamicPropertyFactory.getInstance().getStringProperty("hystrix.monitor.url", "");
    private DynamicBooleanProperty concurrentThreadsNotice = DynamicPropertyFactory.getInstance().getBooleanProperty("concurrent.threads.notice", false);
    private DynamicStringProperty concurrentThreadsNoticeEnvs = DynamicPropertyFactory.getInstance().getStringProperty("concurrent.threads.notice.envs", "");
    private DynamicStringProperty concurrentThreadsNoticeUrls = DynamicPropertyFactory.getInstance().getStringProperty("concurrent.threads.notice.urls", "");
    //监控频率
    private DynamicIntProperty monitoringFrequency = DynamicPropertyFactory.getInstance().getIntProperty("concurrent.threads.notice.monitoring.frequency", 100);

    @Async
    @Scheduled(cron = "0/10 * * * * ?")
    public void noticeTask() {

        this.doSchedule();

    }

    private void doSchedule() {
        if (!concurrentThreadsNotice.get() || !Detect.notEmpty(concurrentThreadsNoticeEnvs.get())
                || !Detect.notEmpty(concurrentThreadsNoticeUrls.get()) || !Detect.notEmpty(hystrixMonitorUrl.get())) {
            return;
        }
        List<String> envs = Arrays.asList(concurrentThreadsNoticeEnvs.get().split(","));
        List<String> urls = Arrays.asList(concurrentThreadsNoticeUrls.get().split(","));

        if (envs.size() != urls.size()) {
            return;
        }

        List<ServiceGateway> serviceGateways = serviceGatewayDao.getAll();

        if (!Detect.notEmpty(serviceGateways)) {
            return;
        }


        List<String> clusters = Lists.newArrayList();

        serviceGateways.forEach(e -> {
            if (Detect.notEmpty(e.getHystrixUrl()) && envs.contains(e.getEnv())) {

                String cluster = e.getName() + "_" + e.getEnv();
                clusters.add(cluster);
            }
        });

        if (!Detect.notEmpty(clusters)) {
            return;
        }

        this.dealNotice(clusters);
    }

    private void dealNotice(List<String> clusters) {
        clusters.forEach(e -> {
            Map<String, JSONObject> resultMap = null;
            try {
                resultMap = sendGet(hystrixMonitorUrl.get(), "cluster=" + e);
            } catch (Exception e1) {
                LOGGER.warn("[调用用监控地址异常:{}?cluster={}]", hystrixMonitorUrl.get(), e);
                return;
            }

            try {
                DataSourceChooseSupportHolder.set(e);
                this.dealNotice(e, resultMap);
            } finally {
                DataSourceChooseSupportHolder.reset();
            }
        });
    }

    private void dealNotice(String cluster, Map<String, JSONObject> resultMap) {
        if (!Detect.notEmpty(resultMap)) {
            return;
        }

        List<String> serviceIds = Lists.newArrayList(resultMap.keySet());

        //去除对应的路由版本
        RouteVersionQuery routeVersionQuery = new RouteVersionQuery();
        routeVersionQuery.setServiceIds(serviceIds);
        List<RouteVersionEnhanced> routeVersionEnhanceds = routeVersionService.getPagination(routeVersionQuery, new Pagination(1, Integer.MAX_VALUE, false), null).getItems();

        RouteQuery routeQuery = new RouteQuery();
        routeQuery.setServiceIds(serviceIds);

        List<RouteEnhanced> routeEnhanceds = routeService.getPagination(routeQuery, new Pagination(1, Integer.MAX_VALUE, false), null).getItems();

        List<Map<String, Object>> noticeDatas = Lists.newArrayList();


        resultMap.entrySet().forEach(e -> {
            routeEnhanceds.forEach(r -> {
                if (e.getKey().equals(r.getRouteServiceEnhanced().getId()) && e.getValue().getLong("propertyValue_metricsRollingStatisticalWindowInMilliseconds") > 0) {
                    long curRequestsPerSec = e.getValue().getInteger("requestCount") / (e.getValue().getLong("propertyValue_metricsRollingStatisticalWindowInMilliseconds") / 1000);

                    if (curRequestsPerSec / monitoringFrequency.get() > 0) {
                        Map<String, Object> noticeDataMap = Maps.newHashMap();
                        noticeDataMap.put("gateway", cluster.substring(0, cluster.lastIndexOf("_")));
                        noticeDataMap.put("env", cluster.substring(cluster.lastIndexOf("_") + 1));
                        noticeDataMap.put("route", r.getId());
                        noticeDataMap.put("version", "");
                        noticeDataMap.put("curRequestsPerSec", curRequestsPerSec);

                        noticeDatas.add(noticeDataMap);
                    }
                }
            });
        });


        resultMap.entrySet().forEach(e -> {
            routeVersionEnhanceds.forEach(v -> {
                if (e.getKey().equals(v.getRouteServiceEnhanced().getId()) && e.getValue().getLong("propertyValue_metricsRollingStatisticalWindowInMilliseconds") > 0) {
                    long curRequestsPerSec = e.getValue().getInteger("requestCount") / (e.getValue().getLong("propertyValue_metricsRollingStatisticalWindowInMilliseconds") / 1000);

                    if (curRequestsPerSec / monitoringFrequency.get() > 0) {
                        Map<String, Object> noticeDataMap = Maps.newHashMap();
                        noticeDataMap.put("gateway", cluster.substring(0, cluster.lastIndexOf("_")));
                        noticeDataMap.put("env", cluster.substring(cluster.lastIndexOf("_") + 1));
                        noticeDataMap.put("route", v.getRouteEnhanced().getId());
                        noticeDataMap.put("version", v.getVersion());
                        noticeDataMap.put("curRequestsPerSec", curRequestsPerSec);

                        noticeDatas.add(noticeDataMap);
                    }
                }
            });
        });

        this.dealNotice(cluster, noticeDatas);
    }

    private void dealNotice(String cluster, List<Map<String, Object>> noticeDatas) {
        if (!Detect.notEmpty(noticeDatas)) {
            return;
        }

        String noticeUrl = this.getNoticeUrl(cluster);
        RequestData requestData = new RequestData();
        requestData.setRequest(new BaseAMO() {
            private List<Map<String, Object>> notices = noticeDatas;
            private String zUser = "zGateway";
            private String zPwd = "23277f0284c247daa0feff69ad9880ef";

            public String getzUser() {
                return zUser;
            }

            public void setzUser(String zUser) {
                this.zUser = zUser;
            }

            public String getzPwd() {
                return zPwd;
            }

            public void setzPwd(String zPwd) {
                this.zPwd = zPwd;
            }

            public List<Map<String, Object>> getNotices() {
                return notices;
            }

            public void setNotices(List<Map<String, Object>> notices) {
                this.notices = notices;
            }
        });

        LOGGER.info("{}--{}", cluster, JSONObject.toJSONString(requestData));
        //发送通知
        HttpUtility.doPostJson(noticeUrl, JSONObject.toJSONString(requestData));
    }

    private String getNoticeUrl(String cluster) {
        List<String> envs = Arrays.asList(concurrentThreadsNoticeEnvs.get().split(","));
        List<String> urls = Arrays.asList(concurrentThreadsNoticeUrls.get().split(","));

        for (int i = 0; i < envs.size(); i++) {
            if (cluster.endsWith(envs.get(i))) {
                return urls.get(i);
            }
        }

        return null;
    }


    public static Map<String, JSONObject> sendGet(String url, String param) throws Exception {
        BufferedReader in = null;
        Map<String, JSONObject> resultMap = Maps.newHashMap();

        try {
            String urlNameString = url;
            if (Detect.notEmpty(param)) {
                urlNameString = urlNameString + "?" + param;
            }
            URL realUrl = new URL(urlNameString);
            // 打开和URL之间的连接
            URLConnection connection = realUrl.openConnection();
            // 设置通用的请求属性
            connection.setRequestProperty("accept", "*/*");
            connection.setRequestProperty("connection", "Keep-Alive");
            connection.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
            connection.setConnectTimeout(3000);// 链接超时(3秒)
            connection.setReadTimeout(5000);// 读取超时(5秒)
            // 建立实际的连接
            connection.connect();

            // 定义 BufferedReader输入流来读取URL的响应
            in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
            int pingCount = 0;//ping count
            String line;
            while ((line = in.readLine()) != null) {
                if (line.startsWith(": ping")) {
                    pingCount++;
                    continue;
                }

                if (pingCount > 1) {
                    break;
                }

                if (line.startsWith("data:")) {
                    line = line.substring(line.indexOf(":") + 1);

                    JSONObject dataJson = JSONObject.parseObject(line);
                    if ("HystrixCommand".equals(dataJson.getString("type"))) {
                        String serviceId = dataJson.getString("name");
                        resultMap.put(serviceId, dataJson);
                    }
                }
            }
        } catch (Exception e) {
            throw new Exception("HTTP请求失败：" + url, e);
        }
        // 使用finally块来关闭输入流
        finally {
            try {
                if (in != null) {
                    in.close();
                }
            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }
        return resultMap;
    }
}
